/**
 * 标量函数
 */
package com.atguigu.day8

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.types.Row

object ScalarFunctionExample {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val settings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()



    val tEnv = StreamTableEnvironment.create(env,settings)
    val stream = env.addSource(new SensorSource)
    val hashCode = new HashCode(10)

    tEnv.registerFunction("hashCode",hashCode)

    val table1 = tEnv.fromDataStream(stream)

    //table写法
    table1
      .select('id,hashCode('id))
      .toAppendStream[Row]
//      .print()


    //sql写法
    tEnv.registerFunction("hashCode",hashCode)
    tEnv.createTemporaryView("sensor",table1)

    tEnv
        .sqlQuery("SELECT id,hashCode(id) FROM sensor")
        .toAppendStream[Row]
        .print()

    env.execute()




  }

  class HashCode(factor:Int) extends ScalarFunction{
    def eval(s:String): Int ={
      s.hashCode+factor
    }

  }
}
